package com.atguigu.medical.realtime.app.dwd;

import com.atguigu.medical.realtime.app.BaseSQLApp;
import com.atguigu.medical.realtime.common.Constant;
import com.atguigu.medical.realtime.util.SQLUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @Author lzc
 * @Date 2023/5/26 19:43
 * 诊断域,医生问诊患者
 */
public class DwdDiagnosisConsultation extends BaseSQLApp {
    public static void main(String[] args) {
        new DwdDiagnosisConsultation().init(
            30002,
            2,
            "DwdDiagnosisConsultation"
        );
        
    }
    
    @Override
    protected void handle(StreamExecutionEnvironment env,
                          StreamTableEnvironment tEnv) {
        // 1. 读取 ods_medical 表
        readOdsMedical(tEnv, "DwdInteractionReview");
        
        // 2. 过滤出评价数据
        Table result = tEnv
            .sqlQuery("select " +
                          " after['id'] id,  " +
                          " after['create_time'] consultation_time,  " +
                          " after['consultation_fee'] consultation_fee,  " +
                          " after['diagnosis'] diagnosis,  " +
                          " after['doctor_id'] doctor_id,  " +
                          " after['patient_id'] patient_id,  " +
                          " after['user_id'] user_id,  " +
                          " ts " +
                          "from ods_medical " +
                          "where source['db']='medical' " +
                          "and source['table']='consultation' " +
                          "and (" +
                          " (op='c' and after['diagnosis'] is not null) " +
                          " or (op='u' and before['diagnosis'] is null and after['diagnosis'] is not null)" +
                          ")");
        // 3. 写出到 kafka
        tEnv.executeSql("create table dwd_diagnosis_consultation(" +
                            " id string,  " +
                            " consultation_time string,  " +
                            " consultation_fee string,  " +
                            " diagnosis string,  " +
                            " doctor_id string,  " +
                            " patient_id string,  " +
                            " user_id string,  " +
                            " ts bigint " +
                            ")" + SQLUtil.getKafkaSinkDDL(Constant.TOPIC_DWD_DIAGNOSIS_CONSULTATION));
        result.executeInsert("dwd_diagnosis_consultation");
    }
}
